package com.uestc.coprocessor.userfeedback;

import java.io.IOException;
import java.util.HashMap;
import java.util.LinkedHashMap;
import java.util.Map;

import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.client.coprocessor.Batch;
import org.apache.hadoop.hbase.ipc.BlockingRpcCallback;

import com.google.protobuf.ServiceException;
import com.uestc.coprocessor.userfeedback.UserFeedbackProtocol.UserFeedbackService;
import com.uestc.util.JsonUtil;

public class UserFeedbackClient {
	public String getCallback(Table table, String type) throws ServiceException, Throwable {
		if (table == null || type == null || type.isEmpty())
			return null;
		final UserFeedbackProtocol.CountRequest request = UserFeedbackProtocol.CountRequest.newBuilder().setType(type)
				.build();
		Map<byte[], String> resultFromRegionServers = table.coprocessorService(
				UserFeedbackProtocol.UserFeedbackService.class, null, null,
				new Batch.Call<UserFeedbackProtocol.UserFeedbackService, String>() {

					@Override
					public String call(UserFeedbackService service) throws IOException {
						BlockingRpcCallback<UserFeedbackProtocol.CountResponse> rpcCallback = new BlockingRpcCallback<>();
						service.getUserFeedback(null, request, rpcCallback);
						UserFeedbackProtocol.CountResponse response = rpcCallback.get();
						String result = response.getResult();
						return result;
					}
				});

		Map<String, Map<String, Long>> item = new LinkedHashMap<>();
		for (String s : resultFromRegionServers.values()) {
			if (s == null || s.isEmpty())
				break;
			String[] trueAndFalse = s.split("&");
			if (trueAndFalse.length < 2)
				break;
			String trueQuestions = trueAndFalse[0];
			String falseQuestions = trueAndFalse[1];
			String[] trueResults = trueQuestions.split(";");
			for (String r : trueResults) {
				String[] ss = r.split(":");
				if (ss.length < 2)
					continue;
				Map<String, Long> value = item.get(ss[0]);
				if (value == null) {
					value = new HashMap<>();
					value.put("true", Long.valueOf(ss[1]));
					value.put("false", 0l);
					item.put(ss[0], value);
				} else {
					Long l = value.get("true");
					value.put("true", l + Long.valueOf(ss[1]));
				}
			}
			String[] falseResults = falseQuestions.split(";");
			for (String f : falseResults) {
				String[] ss = f.split(":");
				if (ss.length < 2)
					continue;
				Map<String, Long> value = item.get(ss[0]);
				if (value == null) {
					value = new HashMap<>();
					value.put("true", 0l);
					value.put("false", Long.valueOf(ss[1]));
					item.put(ss[0], value);
				} else {
					Long l = value.get("false");
					value.put("false", l + Long.valueOf(ss[1]));
				}
			}
		}

		String j1 = JsonUtil.createJsonString(item);
		return j1;
	}
}
